package com.explorex.infra.puma.gateway.netty.ext;

import com.alibaba.fastjson.JSON;
import com.explorex.infra.puma.gateway.netty.common.*;
import com.explorex.infra.puma.gateway.netty.common.ack.AckMessageManager;
import com.explorex.infra.puma.gateway.netty.common.enums.QosEnum;
import com.explorex.infra.puma.gateway.netty.common.utils.MessageBuilderUtils;
import com.explorex.infra.puma.gateway.netty.ws.WebSocketSession;
import com.explorex.infra.puma.gateway.netty.ws.message.Message;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author: JacceYang
 * @createTime: 2022/07/19 8:49 下午
 * @description: 对外的消息发射器，服务端主动发起请求时调用
 */
@Slf4j
public class MessageEmitter {
    private static final ExecutorService threadPoolExecutor =  Executors.newCachedThreadPool();

    @Setter
    @Getter
    private static AbstractBaseServer baseServer;



    private MessageEmitter() {
    }

    public static int emitMessage(String senderId,String domain, int bizId, Set<String> userIds, Object msg, int qos, int contentType, long startTime) {
        if (baseServer == null) {
            return -1;
        }

            List<BaseSocketSession> webSocketSessions = ChannelManager.getInstance().getAllSocketSession();
            if (CollectionUtils.isEmpty(webSocketSessions)) {
                return 0;
            }

            threadPoolExecutor.submit(() -> {
                int result = 0;
                boolean send = false;
                for (BaseSocketSession webSocketSession : webSocketSessions) {
                    if(Objects.equals(senderId,webSocketSession.userId())){
                        continue;
                    }
                    result += sendWebSocketMessage((WebSocketSession) webSocketSession, domain, bizId, msg, contentType, startTime, qos);
                    // 一个 用户有一个端发送成功了，就看作发送成功
                    if (result == 0 && !send) {
                        send = true;
                    }
                }
            });

        return 0;
    }

    private static int sendWebSocketMessage(WebSocketSession webSocketSession, String domain, int bizId, Object msg, int contentType, long startTime, int qos) {
        Throwable throwable = null;
        int result = 0;
        // 通道太拥挤了,低优先消息可以直接丢弃啦
        if (webSocketSession.isRateLimited() && qos == QosEnum.MOST_ONCE.getValue()) {
            return 0;
        }
        Message message = MessageBuilderUtils.buildSendMessage(domain, bizId, contentType, webSocketSession.nextSeqId(), msg, qos);
        try {
            webSocketSession.sendMessage(message);
//            AckMessageManager.sendMessage(webSocketSession
//                    , new WaitingAckMessage(bizId, message, message.getHeader().getSeq(), System.currentTimeMillis()));
        } catch (Exception e) {
            result = -1;
            log.error("write message failed for Exception,contentTypeStr={}", contentType, e);
            throwable = e;
        } finally {
        }
        return result;
    }


}
